package com.afdor.rws.event.driven.support.channel.producer;

import com.afdor.rws.core.exception.BusinessException;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.core.strategy.IdGeneratorStrategy;
import com.afdor.rws.event.driven.api.Serialization;
import com.afdor.rws.event.driven.api.channel.producer.EventProducer;
import com.afdor.rws.event.driven.api.channel.producer.transaction.EventProducerTransactionItem;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.support.channel.producer.transaction.impl.DefaultEventProducerTransactionItemImpl;
import com.afdor.rws.event.driven.support.holder.CallbackEvent;
import com.afdor.rws.event.driven.support.holder.EventResult;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 生产者抽象实现
 *
 * @author 悭梵
 * @date Created in 2018-12-27 9:34
 */
@Slf4j
public abstract class AbstractEventProducer<PC extends EventProducerConfig> implements EventProducer<PC> {

    /**
     * 初始化状态
     */
    private AtomicBoolean init = new AtomicBoolean(Boolean.FALSE);

    /**
     * 配置信息
     */
    @Getter
    @NonNull
    private PC config;
    /**
     * 对应事件源
     */
    @Getter
    @NonNull
    private String eventSourceName;
    /**
     * 生产者名称
     */
    @Getter
    @NonNull
    private String producerName;
    /**
     * 序列化方式
     */
    @Getter
    @NonNull
    private Serialization serialization;

    /**
     * 事物提交队列管理器
     */
    @Getter
    private ConcurrentMap<String, EventProducerTransactionItem> eventProducerTransactionItemMap = new ConcurrentHashMap<String, EventProducerTransactionItem>(64);

    @Override
    public final <T extends EventProducer> T init() {
        checkup(Boolean.FALSE);
        doInit();
        this.init.compareAndSet(Boolean.FALSE, Boolean.TRUE);
        return (T) this;
    }

    @Override
    public synchronized <T extends EventProducer> T setConfig(@NonNull PC config) {
        if (StringUtils.isBlank(config.getProducerName())) {
            throw new NullPointerException("生产者名称为空");
        }
        if (StringUtils.isBlank(config.getEventSourceName())) {
            throw new NullPointerException("事件源名称为空");
        }
        if (StringUtils.isBlank(config.getSerializationStrategyName())) {
            throw new NullPointerException("事件序列化策略名称为空");
        }

        this.config = config;
        this.producerName = this.config.getProducerName();
        this.eventSourceName = this.config.getEventSourceName();
        this.serialization = ExtensionLoader.getInstance(Serialization.class)
                .getExtensionOrDefault(this.config.getSerializationStrategyName());

        if (this.serialization == null) {
            throw new NullPointerException(this.config.getSerializationStrategyName() + "对应事件序列化策略不存在");
        }
        return (T) this;
    }

    @Override
    public EventProducerTransactionItem newTransaction() {
        checkup();
        ExtensionLoader<IdGeneratorStrategy> idGeneratorStrategyExtensionLoader = ExtensionLoader.getInstance(IdGeneratorStrategy.class);
        // 获取配置的Id生成策略
        String transactionId = String.valueOf(idGeneratorStrategyExtensionLoader.getExtensionOrDefault(getConfig().getTransactionStrategyName()).generate());
        if (getEventProducerTransactionItemMap().containsKey(transactionId)) {
            return newTransaction();
        } else {
            getEventProducerTransactionItemMap().putIfAbsent(transactionId, createEventProducerTransactionItem(transactionId));
        }
        return getEventProducerTransactionItemMap().get(transactionId);
    }

    @Override
    public void addEvent(@NonNull String transactionId, @NonNull Event event) {
        checkup();
        getEventProducerTransactionItem(transactionId).addEvent(event);
    }

    @Override
    public boolean commit(@NonNull String transactionId) {
        checkup();
        try {
            return getEventProducerTransactionItem(transactionId).commit();
        } finally {
            cleanEventProducerTransactionItem(transactionId);
        }
    }

    @Override
    public void rollback(@NonNull String transactionId) {
        checkup();
        try {
            getEventProducerTransactionItem(transactionId).rollback();
        } finally {
            cleanEventProducerTransactionItem(transactionId);
        }
    }

    @Override
    public final void destroy() {
        this.init.compareAndSet(Boolean.TRUE, Boolean.FALSE);
        /**
         * 未提交事物全部回滚
         */
        getEventProducerTransactionItemMap()
                .entrySet()
                .parallelStream()
                .forEach(entry -> entry.getValue().rollback());

        getEventProducerTransactionItemMap().clear();

        doDestroy();
    }

    /**
     * 通过生产者，发送单条事件
     *
     * @param event 事件对象
     */
    @Override
    public <V> EventResult<V> sendEvent(@NonNull Event event) {
        throw new BusinessException("不支持同步消息发送");
    }

    /**
     * 异步发送消息并成功回调同步处理
     *
     * @param callbackEvent
     */
    @Override
    public <V> void asyncSendEvent(@NonNull CallbackEvent<V> callbackEvent) {
        throw new BusinessException("不支持异步回调消息发送");
    }

    /**
     * 发送异步消息
     *
     * @param event
     * @return
     * @throws Exception
     */
    @Override
    public void sendAsyncEvent(@NonNull Event event) {
        throw new BusinessException("不支持异步消息发送");
    }

    /**
     * 批量发送异步消息
     *
     * @param eventList
     * @return
     */
    @Override
    public void batchSendAsyncEvent(@NonNull List<Event> eventList) {
        throw new BusinessException("不支持批量异步消息发送");
    }


    /**
     * 启动
     */
    protected void doInit() {

    }

    /**
     * 终止
     */
    protected void doDestroy() {

    }

    /**
     * 创建事物管理器
     */
    protected EventProducerTransactionItem createEventProducerTransactionItem(@NonNull String transactionId) {
        return new DefaultEventProducerTransactionItemImpl(transactionId, this);
    }

    /**
     * 是否初始化
     *
     * @return
     */
    protected boolean isInit() {
        return this.init.get();
    }

    /**
     * 验证
     */
    protected void checkup() {
        checkup(Boolean.TRUE);
    }

    /**
     * 验证
     */
    private void checkup(boolean vaildInit) {
        if (vaildInit) {
            if (!isInit()) {
                throw new RuntimeException("请先执行初始化相关操作");
            }
        } else {
            if (Objects.isNull(getConfig())) {
                throw new NullPointerException("生产者相关配置信息为空");
            }
            if (Objects.isNull(getSerialization())) {
                throw new RuntimeException("事件序列化处理器为空");
            }
            if (StringUtils.isBlank(getEventSourceName())) {
                throw new RuntimeException("事件源名称为空");
            }
        }
    }

    /**
     * 获取事物项
     *
     * @param transactionId
     * @return
     */
    private EventProducerTransactionItem getEventProducerTransactionItem(@NonNull String transactionId) {
        EventProducerTransactionItem transactionItem = getEventProducerTransactionItemMap().get(transactionId);
        if (transactionItem == null) {
            throw new IllegalStateException("the specified transaction is not turned on or has been executed. please open a new transaction first . ");
        }
        return transactionItem;
    }

    /**
     * 清理事务项信息
     *
     * @param transactionId
     */
    private void cleanEventProducerTransactionItem(@NonNull String transactionId) {
        getEventProducerTransactionItemMap().remove(transactionId);
    }
}
